package com.lgw.coupon.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.json.JSONUtil;
import com.lgw.commons.constant.ApiConstant;
import com.lgw.commons.constant.KafkaConstant;
import com.lgw.commons.constant.RedisKeyConstant;
import com.lgw.commons.exception.ParameterException;
import com.lgw.commons.model.domain.ResultInfo;
import com.lgw.commons.model.domain.SignInIdentity;
import com.lgw.commons.model.pojo.SeckillVouchers;
import com.lgw.commons.model.pojo.VoucherOrders;
import com.lgw.commons.utils.AssertUtil;
import com.lgw.commons.utils.ResultInfoUtil;
import com.lgw.commons.utils.ScriptUtil;
import com.lgw.coupon.mapper.VoucherOrdersMapper;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import com.lgw.coupon.service.ISeckillService;
import org.springframework.transaction.interceptor.TransactionAspectSupport;
import org.springframework.web.client.RestTemplate;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class SeckillServiceImpl implements ISeckillService {

    @Autowired
    private RedisTemplate redisTemplate;
    @Value("${service.name.oauth-server}")
    private String oauthServerName;
    @Autowired
    private RestTemplate restTemplate;

    private DefaultRedisScript<Long> stockScript;

    @Autowired
    private KafkaTemplate kafkaTemplate;
    @Resource
    private VoucherOrdersMapper voucherOrdersMapper;
    //Kafka客户端
//    @Autowired
//    private KafkaTemplate<String, String> kafkaTemplate;
    @Resource
    private RedissonClient redissonClient;

    //读取lua脚本
    @PostConstruct
    public void init() {
        stockScript = new DefaultRedisScript<>();
        stockScript.setLocation(new ClassPathResource("stock.lua"));
        stockScript.setResultType(Long.class);
    }

    /**
     * 添加秒杀活动
     *
     * @param seckillVouchers
     */
    @Transactional(rollbackFor = Exception.class)
    @Override
    public void addSeckillVouchers(SeckillVouchers seckillVouchers) {
        //校验
        AssertUtil.isTrue(seckillVouchers.getVoucherId() == null, "请选择需要抢购的优惠券");
        AssertUtil.isTrue(seckillVouchers.getAmount() == 0, "请输入抢购总数量");
        Date cur = new Date();
        AssertUtil.isNotNull(seckillVouchers.getStartTime(), "请输入开始时间");
        AssertUtil.isNotNull(seckillVouchers.getEndTime(), "请输入结束时间");
//        AssertUtil.isTrue(cur.after(seckillVouchers.getStartTime()), "开始时间不能早于当前时间");
        AssertUtil.isTrue(seckillVouchers.getStartTime().after(seckillVouchers.getEndTime()), "开始时间不能晚于结束时间");
        //生成redis抢购码
        String key = RedisKeyConstant.seckill_vouchers.getKey() + seckillVouchers.getVoucherId();
        //验证是否存在该券的秒杀活动
        Map<String, Object> map = redisTemplate.opsForHash().entries(key);
        AssertUtil.isTrue(!map.isEmpty() && (int) map.get("amount") > 0, "该券已经拥有抢购活动");
        //插入redis
        seckillVouchers.setIsValid(1);
        seckillVouchers.setCreateDate(cur);
        seckillVouchers.setUpdateDate(cur);
        redisTemplate.opsForHash().putAll(key, BeanUtil.beanToMap(seckillVouchers));
        //使用库存作为Redisson信号量限制库存

    }

    /**
     * 商品秒杀，限制一人一单，防止超卖
     *
     * @param voucherId
     * @param accessToken
     * @param path
     * @return
     */
    @Transactional(rollbackFor = Exception.class)
    @Override
    public ResultInfo doSeckill(Integer voucherId, String accessToken, String path) {
        //校验
        AssertUtil.isTrue(voucherId == null || voucherId < 0, "请选择需要抢购的代金券");
        AssertUtil.isNotEmpty(accessToken, "请登录");
        //获取rediskey,获取活动信息
        String redisKey = RedisKeyConstant.seckill_vouchers.getKey() + voucherId;
        Map<String, Object> map = redisTemplate.opsForHash().entries(redisKey);
        SeckillVouchers seckillVouchers = BeanUtil.mapToBean(map, SeckillVouchers.class, true, null);
        //判断活动是否开始，结束
        Date cur = new Date();
        AssertUtil.isTrue(seckillVouchers.getStartTime().after(cur), "抢购活动还未开始");
        AssertUtil.isTrue(cur.after(seckillVouchers.getEndTime()), "抢购活动已结束");
        //判断是否已经卖完
        AssertUtil.isTrue(seckillVouchers.getAmount() < 1, "该券已经卖完了");
        //获取登录用户
        String url = oauthServerName + "user/me?access_token={accesstoken}";
        ResultInfo resultInfo = restTemplate.getForObject(url, ResultInfo.class, accessToken);
        if (resultInfo.getCode() != ApiConstant.SUCCESS_CODE) {
            resultInfo.setPath(path);
            return resultInfo;
        }
        //注意data的数据是LinkHashMap类型
        SignInIdentity customerInfo = BeanUtil.fillBeanWithMap((LinkedHashMap) resultInfo.getData(),
                new SignInIdentity(), false);
        //判断用户是否已经抢到(只能买一件)
        VoucherOrders order = voucherOrdersMapper.findCustomerOrder(customerInfo.getId(),
                seckillVouchers.getVoucherId());
        //使用Redis锁 一个账号只能买一次
        AssertUtil.isTrue(order != null, "该用户已抢到代金券，无需再抢");
        String lockName = RedisKeyConstant.lock_key.getKey() + customerInfo.getId() + ":" + voucherId;
        //加锁
        long expireTime = seckillVouchers.getEndTime().getTime() - cur.getTime();
        //限制一人一单 设置锁定资源名称
        //RLock lock = redissonClient.getLock(lockName);

        try {
            // Redisson 分布式锁处理
//            boolean isLocked = lock.tryLock(expireTime, TimeUnit.MILLISECONDS);
            Integer num = 1;
            boolean isLocked = redisTemplate.opsForValue().setIfAbsent(lockName, num.toString(), expireTime, TimeUnit.MILLISECONDS);
            if (isLocked) {
                List<String> keys = new ArrayList<>();
                keys.add(redisKey);
                keys.add("amount");
                Long amount = (Long) redisTemplate.execute(stockScript, keys);
                log.info(String.valueOf(amount));
                AssertUtil.isTrue(amount == null || amount < 1, "该券已经卖完了");

                VoucherOrders voucherOrders = new VoucherOrders();
                voucherOrders.setCustomerId(customerInfo.getId());
                voucherOrders.setVoucherId(seckillVouchers.getVoucherId());
                String orderNo = IdUtil.getSnowflake(1, 1).nextIdStr();
                voucherOrders.setOrderNo(orderNo);
                voucherOrders.setStatus(0);
//                long count = voucherOrdersMapper.save(voucherOrders);
//                log.info("下单：{}",count);
//                AssertUtil.isTrue(count == 0, "用户抢购失败");
                try {
                    kafkaTemplate.send(KafkaConstant.TOPIC, JSONUtil.toJsonStr(voucherOrders));
//                AssertUtil.isTrue(count == 0, "用户抢购失败");
                } catch (Exception e) {
                    log.info("kafka推送出现问题");
                    redisTemplate.opsForHash().increment(redisKey,"amount",1);
                    e.printStackTrace();
                    throw(e);
                }

            }
        } catch (Exception e) {
            // 手动回滚事务
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
            log.error("事务回滚");
            // Redisson 解锁
//            lock.unlock();
            redisTemplate.delete(lockName);

            if (e instanceof ParameterException) {
                return ResultInfoUtil.buildError(0, "该券已经卖完了", path);
            }
            return ResultInfoUtil.buildError(0, "抢购失败",path);
        }
        return ResultInfoUtil.buildSuccess(path, "抢购成功");

    }
}

//        Integer num = 1;
////        boolean isLocked = redisTemplate.opsForValue().setIfAbsent(lockName, num.toString(), expireTime, TimeUnit.MILLISECONDS);
//        if (isLocked) {
//            //占位成功，说明该用户未秒杀过该商品，则继续
//            List<String> keys = new ArrayList<>();
//            keys.add(redisKey);
//            keys.add("amount");
//            //stockScript是lua脚本，先判定库存是否够再减，
//            Long amount = (Long) redisTemplate.execute(stockScript, keys);
//            if (amount == null || amount < 1) {
//                redisTemplate.delete(lockName);
//                return ResultInfoUtil.buildError(0, "该券已经卖完了", path);
//            }
////            AssertUtil.isTrue(amount == null || amount < 1, "该券已经卖完了");
//            //发送到kafka进行异步下单
//            VoucherOrders voucherOrders = new VoucherOrders();
//            voucherOrders.setCustomerId(customerInfo.getId());
//            voucherOrders.setVoucherId(seckillVouchers.getVoucherId());
//            String orderNo = IdUtil.getSnowflake(1, 1).nextIdStr();
//            voucherOrders.setOrderNo(orderNo);
//            voucherOrders.setStatus(0);
//            try {
//                kafkaTemplate.send(KafkaConstant.TOPIC, JSONUtil.toJsonStr(voucherOrders));
//            } catch (Exception e) {
////                TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
//
//                redisTemplate.opsForHash().increment(redisKey, "amount", 1);
//                redisTemplate.delete(lockName);
////                redisTemplate.opsForHash().increment(redisKey, "amount", 1);
//                e.printStackTrace();
//                log.error("下单出现异常");
//            }
//        } else {
//            return ResultInfoUtil.buildError(0, "用户购买过", path);
//        }
//        // 手动回滚事务
////            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
//        // 占位符 解锁
////        redisTemplate.delete(lockName);
////      redisTemplate.opsForHash().increment(redisKey,"amount",1);
//        return ResultInfoUtil.buildSuccess(path, "抢购成功");
//    }
//}
